/**
 * 增量聚合函数
 */

package com.atguigu.stream.window

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object AggregateFunction {
  def main(args: Array[String]): Unit = {



    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream = env.addSource(new SensorSource)
    stream.keyBy(_.id)
      .timeWindow(Time.seconds(10))
      .aggregate(new AvgTemoAgg)
      .print()

    env.execute()

  }


  //第一个泛型：流中元素类型
  //第二个泛型：元组(传感器id，来了多少条温度，温度总和
  //第三个泛型：返回值类型，
  class AvgTemoAgg extends AggregateFunction[SensorReading,(String,Long,Double),(String,Double)]{
//    创建空累加器
    override def createAccumulator(): (String, Long, Double) =("",0L,0.0)
//    聚合逻辑
    override def add(in: SensorReading, acc: (String, Long, Double)): (String, Long, Double) = {
      (in.id,acc._2+1,acc._3+in.temperature)

    }
//    窗口聚合时输出什么
    override def getResult(acc: (String, Long, Double)): (String, Double) = {
      (acc._1,acc._3/acc._2)


    }
//    累加器聚合逻辑是什么
    override def merge(acc: (String, Long, Double), acc1: (String, Long, Double)): (String, Long, Double) = {

      (acc._1,acc._2+acc1._2,acc._3+acc1._3)

    }
  }

}
